package org.ala.tiktools.rpc.kcp.netty.client;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.List;
import java.util.concurrent.TimeUnit;

import org.ala.tiktools.rpc.kcp.config.RpcKcpConfigInfo;
import org.ala.tiktools.rpc.kcp.exception.KcpRpcException;
import org.ala.tiktools.rpc.kcp.netty.IKcpChannel;
import org.ala.tiktools.rpc.kcp.netty.NettyKcp;
import org.ala.tiktools.rpc.kcp.netty.client.config.DefaultKcpClientChannelConfig;
import org.ala.tiktools.rpc.kcp.netty.client.config.IKcpClientChannelConfig;
import org.ala.tiktools.rpc.kcp.netty.tools.NettyTools;
import org.ala.tiktools.rpc.kcp.protocol.IKcpOutput;
import org.ala.tiktools.rpc.kcp.protocol.Kcp;
import org.ala.tiktools.tools.TimestampTools;

import io.netty.buffer.ByteBuf;
import io.netty.channel.AbstractChannel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelMetadata;
import io.netty.channel.ChannelOutboundBuffer;
import io.netty.channel.ChannelPromise;
import io.netty.channel.EventLoop;
import io.netty.channel.nio.NioEventLoop;
import io.netty.util.internal.logging.InternalLogger;
import io.netty.util.internal.logging.InternalLoggerFactory;

/**
 * 1）只负责update定时任务，不负责read逻辑<br/>
 *      update负责flush，发送数据
 * 2）read逻辑在udpChannel.unsafe()中被执行<br/>
 *
 * @author ala
 * @date 2025-04-15 22:39
 */
public class KcpClientChannel extends AbstractChannel implements IKcpChannel, Runnable {
    private static final InternalLogger log = InternalLoggerFactory.getInstance(KcpClientChannel.class);
    /**
     *  配置
     */
    protected IKcpClientChannelConfig config;
    /**
     *
     */
    protected IKcpOutput output = new KcpClientOutput();
    /**
     *
     */
    protected NettyKcp nkcp;
    /**
     *
     */
    protected KcpClientUdpChannel udpChannel;
    /**
     *  是否正在flush
     */
    protected boolean flushPending;
    /**
     *  关闭udpChannel标记位，只有第一次close时会调用udpChannel.close()
     */
    protected boolean closeAnother = false;
    /**
     *  上次update时间戳
     */
    protected int tsUpdate;
    /**
     *  已经设置了update标记位
     */
    protected boolean scheduleUpdate;


    public KcpClientChannel() {
        super(null);
        udpChannel = new KcpClientUdpChannel(this);
        nkcp = createNettyKcp();
        config = new DefaultKcpClientChannelConfig(this, nkcp, udpChannel.javaChannel().socket());
    }
    /**
     *  构建NettyKcp
     */
    protected NettyKcp createNettyKcp() {
        NettyKcp kcp = new NettyKcp(0, output);
        kcp.channel(this);
        kcp.setFastFlush(RpcKcpConfigInfo.INSTANCE.isIkcpFastFlush());
        return kcp;
    }


    /**
     *  是否可读
     *  <p> rcvQue中是否存在完整消息可读
     */
    public boolean kcpCanRecv() {
        return nkcp.canRecv();
    }
    /**
     *  执行receive, rcvQue -> BufBuf
     */
    public void kcpReceive(List<ByteBuf> bufs) {
        nkcp.receive(bufs);
    }
    /**
     *  执行receive, rcvQue -> BufBuf
     */
    public void kcpReceive(ByteBuf buf) {
        nkcp.receive(buf);
    }
    /**
     *  检测rcvQue中，第一个完整消息的字节长度
     */
    public int kcpPeekSize() {
        return nkcp.peekSize();
    }
    /**
     *  buf -> rcvBuf
     */
    public void kcpInput(ByteBuf buf) throws IOException {
        nkcp.input(buf);
    }
    /**
     * 检测kcp的update时间戳
     */
    public int kcpCheck(int current) {
        return nkcp.check(current);
    }
    /**
     * 设置下次update时间戳
     */
    public void kcpTsUpdate(int tsUp) {
        nkcp.setTsUpdate(tsUp);
    }
    /**
     * 取下次update时间戳
     */
    public int kcpTsUpdate() {
        return nkcp.getTsUpdate();
    }
    /**
     *  是否可以发送
     */
    public boolean kcpCanSend() {
        return nkcp.canSend(!flushPending);
    }
    /**
     *  执行一次flush
     */
    public void updateKcp() {
        int current = TimestampTools.milliSeconds32();
        Throwable ex = null;
        try {
            kcpUpdate(current);
        } catch (Exception e) {
            ex = e;
        }

        if (nkcp.getState() == -1 && ex == null) {
            ex = new KcpRpcException(String.format("执行flush之后 state == -1 clientChannel:%s", this));
        }
        if (ex != null) {
            NettyTools.fireExceptionAndClose(this, ex);
        }
    }
    public int kcpUpdate(int current) {
        return nkcp.update(current);
    }
    /**
     *  数据写入sndQue中
     */
    public boolean kcpSend(ByteBuf buf) throws IOException {
        if (nkcp.canSend(true)) {
            nkcp.send(buf);
            return true;
        } else {
            return false;
        }
    }
    /**
     *  注册定时update任务
     */
    public void scheduleUpdate(int tsUp, int current) {
        this.tsUpdate = tsUp;
        this.scheduleUpdate = true;
        int delay = tsUpdate - current;
        eventLoop().schedule(this, delay, TimeUnit.MILLISECONDS);
    }

    /**
     *  给udpChannel绑定eventLoop
     */
    @Override
    protected void doRegister() throws Exception {
        eventLoop().register(udpChannel).addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                if (!future.isSuccess()) {
                    forceClose(future.cause());
                }
            }
        });
    }
    private void forceClose(Throwable t) {
        unsafe().closeForcibly();
        ((ChannelPromise) closeFuture()).trySuccess();
        log.warn("Failed to register an UkcpClientUdpChannel: {}", this, t);
    }

    @Override
    public void run() {
        if (!isActive()) { return; }

        int current = TimestampTools.milliSeconds32();
        int nextTsUpdate = 0;
        boolean nextSchedule = false;
        int tsUp = kcpTsUpdate();
        Throwable exception = null;

        //  如果当前时间戳 >= 上次update时间戳，直接执行update（触发flush）
        if (current >= tsUp) {
            try {
                nextTsUpdate = kcpUpdate(current);
                nextSchedule = true;
            } catch (Exception e) {
                exception = e;
            }

            if (nkcp.getState() == -1 && exception == null) {
                exception = new KcpRpcException(String.format("执行flush之后 state == -1 channel:%s", this));
            }
        }
        //  如果还没到触发时间戳，继续等待执行
        else {
            nextTsUpdate = tsUp;
            nextSchedule = true;
        }

        boolean close = false;
        if (exception != null) {
            close = true;
            nextTsUpdate = 0;
            nextSchedule = false;
        } else if (isFlushPending() && kcpCanSend()) {
            unsafe().forceFlush();
        }

        tsUpdate = nextTsUpdate;
        scheduleUpdate = nextSchedule;
        if (scheduleUpdate) {
            scheduleUpdate(tsUpdate, current);
        }

        if (close) {
            NettyTools.fireExceptionAndClose(this, exception);
        }
    }


    @Override
    protected void doClose() throws Exception {
        nkcp.setClosed(true);
        if (!closeAnother) {
            closeAnother = true;
            udpChannel.unsafe().close(udpChannel.unsafe().voidPromise());
        }
    }


    /**
     *  在channel.flush时，会被调用
     *  <p> 将sndBuf中的数据，通过KcpOutput发送出去
     *  <p> kcpOutput会触发udpChannel的flush，进而通过udpChannel.doWriteMessage()方法真实发送
     */
    @Override
    protected void doWrite(ChannelOutboundBuffer in) throws Exception {
        boolean sent = false;
        while (true) {
            Object o = in.current();
            if (o == null) {
                flushPending = false;
                break;
            }
            try {
                boolean done = false;
                if (kcpSend((ByteBuf) o)) {
                    done = true;
                    sent = true;
                }
                if (done) {
                    in.remove();
                } else {
                    flushPending = true;
                    break;
                }
            } catch (Exception e) {
                throw e;
            }
        }
        if (sent) {
            if (nkcp.isFastFlush()) {
                updateKcp();
            } else {
                kcpTsUpdate(TimestampTools.milliSeconds32());
            }
        }
    }

    @Override
    protected final Object filterOutboundMessage(Object msg) {
        if (msg instanceof ByteBuf) { return msg; }
        throw new UnsupportedOperationException("只接受ByteBuf类型参数");
    }
    @Override
    public KcpClientUnsafe unsafe() { return (KcpClientUnsafe) super.unsafe(); }
    protected boolean doConnect(SocketAddress remoteAddress, SocketAddress localAddress) throws Exception { return udpChannel.doConnect(remoteAddress, localAddress); }
    @Override
    protected KcpClientUnsafe newUnsafe() { return new KcpClientUnsafe(); }
    @Override
    protected boolean isCompatible(EventLoop loop) { return loop instanceof NioEventLoop; }
    @Override
    protected SocketAddress localAddress0() { return udpChannel.localAddress0(); }
    @Override
    protected SocketAddress remoteAddress0() { return udpChannel.remoteAddress(); }
    public InetSocketAddress localAddress() {return (InetSocketAddress) super.localAddress();}
    public InetSocketAddress remoteAddress() {return (InetSocketAddress) super.remoteAddress();}
    @Override
    public IKcpClientChannelConfig config() { return config; }
    @Override
    public boolean isOpen() { return udpChannel.isOpen(); }
    @Override
    public boolean isActive() { return udpChannel.isActive(); }
    @Override
    public ChannelMetadata metadata() { return udpChannel.metadata(); }
    @Override
    public int conv() { return nkcp.getConv(); }
    @Override
    public IKcpChannel conv(int conv) { nkcp.setConv(conv); return this; }
    @Override
    protected void doDisconnect() throws Exception { udpChannel.doDisconnect(); }
    @Override
    protected void doBeginRead() throws Exception { udpChannel.doBeginRead(); }
    @Override
    protected void doBind(SocketAddress localAddress) throws Exception { udpChannel.doBind(localAddress); }


    /**
     *  客户端的output
     */
    private static class KcpClientOutput implements IKcpOutput {
        @Override
        public void out(ByteBuf data, Kcp kcp) {
            KcpClientChannel ukcpChannel = (KcpClientChannel) kcp.getChannel();
            KcpClientUdpChannel udpChannel = ukcpChannel.udpChannel;
            udpChannel.unsafe().write(data, udpChannel.voidPromise());
            //  flush会走udpChannel.onWriteMessage()，在里面走udp口send
            udpChannel.unsafe().flush();
        }
    }

    /**
     * 客户端的unsafe
     */
    final class KcpClientUnsafe extends AbstractUnsafe {
        @Override
        public void connect(SocketAddress remoteAddress, SocketAddress localAddress, ChannelPromise promise) {
            if (!promise.setUncancellable() || !ensureOpen(promise)) {
                return;
            }
            try {
                boolean wasActive = isActive();
                if (doConnect(remoteAddress, localAddress)) {
                    fulfillConnectPromise(promise, wasActive);
                } else {
                    throw new Error();
                }
            } catch (Throwable t) {
                promise.tryFailure(annotateConnectException(t, remoteAddress));
                closeIfClosed();
            }
        }

        private void fulfillConnectPromise(ChannelPromise promise, boolean wasActive) {
            if (promise == null) {
                // Closed via cancellation and the promise has been notified already.
                return;
            }

            // Get the state as trySuccess() may trigger an ChannelFutureListener that will close the Channel.
            // We still need to ensure we call fireChannelActive() in this case.
            boolean active = isActive();

            // trySuccess() will return false if a user cancelled the connection attempt.
            boolean promiseSet = promise.trySuccess();

            // Regardless if the connection attempt was cancelled, channelActive() event should be triggered,
            // because what happened is what happened.
            if (!wasActive && active) {
                pipeline().fireChannelActive();
            }

            // If a user cancelled the connection attempt, close the channel, which is followed by channelInactive().
            if (!promiseSet) {
                close(voidPromise());
            }
        }
        @Override
        protected void flush0() {
            if (isFlushPending()) {
                return;
            }
            super.flush0();
        }
        void forceFlush() {
            super.flush0();
        }
    }


    public boolean isFlushPending() {
        return flushPending;
    }

    public void setFlushPending(boolean flushPending) {
        this.flushPending = flushPending;
    }

    public boolean isCloseAnother() {
        return closeAnother;
    }

    public void setCloseAnother(boolean closeAnother) {
        this.closeAnother = closeAnother;
    }

    public IKcpOutput getOutput() {
        return output;
    }

    public void setOutput(IKcpOutput output) {
        this.output = output;
    }

    public int getTsUpdate() {
        return tsUpdate;
    }

    public void setTsUpdate(int tsUpdate) {
        this.tsUpdate = tsUpdate;
    }

    public boolean isScheduleUpdate() {
        return scheduleUpdate;
    }

    public void setScheduleUpdate(boolean scheduleUpdate) {
        this.scheduleUpdate = scheduleUpdate;
    }



}
